package com.ruoyi.quartz.mq;

import com.founder.exchange.kafka.message.ECMessage;
import com.founder.exchange.tools.ExchangeTools;
import org.apache.commons.io.IOUtils;
import org.springframework.stereotype.Component;

import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.OutputStream;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;

/**
 * 通过offset接收数据
 *
 * @author wang_zhuming
 */
@Component
public class ConsumerOffset {

    //授权码
    private final static String LICENSE_CODE = "S1563352928485f66e37";

    public ConsumerOffset() {
    }

    public List<String> getData() throws Exception{
        String consumerId = ExchangeTools.createConsumer(LICENSE_CODE);
        Long startTime = new Date().getTime();
        //多少毫秒没有收到数据就跳出循环停止接收
        Long limit = 100000L;
        List<String> list = new ArrayList<String>();
        while (true) {
            List<ECMessage> ecMessage = ExchangeTools.recevie(consumerId);
            System.out.println(ecMessage.size());
            //获得数据刷新时间
            if (ecMessage.size() > 0) {
                startTime = new Date().getTime();
            }
            //计算多久没有收到数据超过限制关闭接收端
            Long time = new Date().getTime() - startTime;
            if (time > limit) {
                ExchangeTools.closeConsumer(consumerId);
                break;
            }
            for (ECMessage message : ecMessage) {
                //打印获取的数据
                String msg = message.getMessageBody().toString();
                list.add(msg);
                System.out.println(msg);
            }

        }
        return list;
    }

    /*
     * 注意！！！！
     * 以下代码只作为实例
     * 请根据实际的使用情况进行修改
     */
/*	public static void main(String[] args) {
		//授权码S1530691686872b237f3  S1527127399012609973
		String consumerId=ExchangeTools.createConsumer("PS1511507721809210541");		
		//开启时间
		Long startTime = new Date().getTime();
		//多少毫秒没有收到数据就跳出循环停止接收
		//请根据实际情况修改
		Long limit=100000L;
		File file=new File("data");
		OutputStream out=null;
		try {
			out=new FileOutputStream(file);		
			while(true){
				List<ECMessage> ecMessage=ExchangeTools.recevie(consumerId);
				System.out.println(ecMessage.size());
				//获得数据刷新时间
				if(ecMessage.size()>0){
					startTime = new Date().getTime();
				}
				//计算多久没有收到数据超过限制关闭接收端
				Long time=new Date().getTime()-startTime;
				if(time>limit){
					ExchangeTools.closeConsumer(consumerId);
					break;
				}	
					List<String> list=new ArrayList<String>();
				for(ECMessage message:ecMessage){
					//打印获取的数据
					String msg=message.getMessageBody().toString();	
					list.add(msg);	
					System.out.println(msg);
				}	
							
				IOUtils.writeLines(list, IOUtils.LINE_SEPARATOR_WINDOWS, out, "UTF8");
				out.flush();
				//如果没有消息消费则不提交
				//请根据实际情况修改
				//if(ecMessage.size()>0){
				//	int success=ExchangeTools.commit(consumerId);
				//	if(success>0){
				//		System.out.println("提交成功");
				//	}			
			//	}
					
			}
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} finally {			
			try {
				if(out!=null)out.close();
			} catch (IOException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
		}
		//循环接收数据
						
	}*/
}
